Apache Kafka |
您所在的位置：网站首页 › kafka error connecting to node › Apache Kafka |
|
概述 Spring-Kafka 提供消费重试的机制。当消息消费失败的时候，Spring-Kafka 会通过消费重试机制，重新投递该消息给 Consumer ，让 Consumer 重新消费消息 。 默认情况下，Spring-Kafka 达到配置的重试次数时，【每条消息的失败重试时间，由配置的时间隔决定】Consumer 如果依然消费失败 ，那么该消息就会进入到死信队列。 Spring-Kafka 封装了消费重试和死信队列， 将正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），将存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。 我们在应用中可以对死信队列中的消息进行监控重发，来使得消费者实例再次进行消费，消费端需要做幂等性的处理。 CodePOM依赖 org.springframework.bootgroupId> spring-boot-starter-webartifactId> dependency> org.springframework.kafkagroupId> spring-kafkaartifactId> dependency> org.springframework.bootgroupId> spring-boot-starter-testartifactId> testscope> dependency> junitgroupId> junitartifactId> testscope> dependency> dependencies>配置文件spring: # Kafka 配置项，对应 KafkaProperties 配置类 kafka: bootstrap-servers: 192.168.126.140:9092 # 指定 Kafka Broker 地址，可以设置多个，以逗号分隔 # Kafka Producer 配置项 producer: acks: 1 # 0-不应答。1-leader 应答。all-所有 leader 和 follower 应答。 retries: 3 # 发送失败时，重试发送的次数 key-serializer: org.apache.kafka.common.serialization.StringSerializer # 消息的 key 的序列化 value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化 # Kafka Consumer 配置项 consumer: auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer properties: spring: json: trusted: packages: com.artisan.springkafka.domain # Kafka Consumer Listener 监听器配置 listener: missing-topics-fatal: false # 消费监听接口监听的主题不存在时，默认会报错。所以通过设置为 false ，解决报错 logging: level: org: springframework: kafka: ERROR # spring-kafka apache: kafka: ERROR # kafka配置类首先要写一个配置类，用于处理消费异常 ErrorHandler package com.artisan.springkafka.configuration; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Primary; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.listener.*; import org.springframework.util.backoff.BackOff; import org.springframework.util.backoff.FixedBackOff; /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/2/18 14:32 * @mark: show me the code , change the world */ @Configuration public class KafkaConfiguration { private Logger logger = LoggerFactory.getLogger(getClass()); @Bean @Primary public ErrorHandler kafkaErrorHandler(KafkaTemplate template) { logger.warn("kafkaErrorHandler begin to Handle"); // 创建 DeadLetterPublishingRecoverer 对象 ConsumerRecordRecoverer recoverer = new DeadLetterPublishingRecoverer(template); // 创建 FixedBackOff 对象 设置重试间隔 10秒 次数为 3次 BackOff backOff = new FixedBackOff(10 * 1000L, 3L); // 创建 SeekToCurrentErrorHandler 对象 return new SeekToCurrentErrorHandler(recoverer, backOff); } // @Bean // @Primary // public BatchErrorHandler kafkaBatchErrorHandler() { // // 创建 SeekToCurrentBatchErrorHandler 对象 // SeekToCurrentBatchErrorHandler batchErrorHandler = new SeekToCurrentBatchErrorHandler(); // // 创建 FixedBackOff 对象 // BackOff backOff = new FixedBackOff(10 * 1000L, 3L); // batchErrorHandler.setBackOff(backOff); // // 返回 // return batchErrorHandler; // } }Spring-Kafka 通过实现自定义的 SeekToCurrentErrorHandler ，当 Consumer 消费消息异常的时候，进行拦截处理： 重试小于最大次数时，重新投递该消息给 Consumer重试到达最大次数时，如果Consumer 还是消费失败时，该消息就会发送到死信队列。 死信队列的 命名规则为： 原有 Topic + .DLT 后缀 = 其死信队列的 Topic ConsumerRecordRecoverer recoverer = new DeadLetterPublishingRecoverer(template);创建 DeadLetterPublishingRecoverer 对象，它负责实现，在重试到达最大次数时，Consumer 还是消费失败时，该消息就会发送到死信队列。 BackOff backOff = new FixedBackOff(10 * 1000L, 3L);也可以选择 BackOff 的另一个子类 ExponentialBackOff 实现，提供指数递增的间隔时间 new SeekToCurrentErrorHandler(recoverer, backOff);创建 SeekToCurrentErrorHandler 对象，负责处理异常，串联整个消费重试的整个过程。 SeekToCurrentErrorHandler在消息消费失败时，SeekToCurrentErrorHandler 会将 调用 Kafka Consumer 的 seek(TopicPartition partition, long offset) 方法，将 Consumer 对于该消息对应的 TopicPartition 分区的本地进度设置成该消息的位置。 这样，Consumer 在下次从 Kafka Broker 拉取消息的时候，又能重新拉取到这条消费失败的消息，并且是第一条。 同时，Spring-Kafka 使用 FailedRecordTracker 对每个 Topic 的每个 TopicPartition 消费失败次数进行计数，这样相当于对该 TopicPartition 的第一条消费失败的消息的消费失败次数进行计数。 另外，在 FailedRecordTracker 中，会调用 BackOff 来进行计算，该消息的下一次重新消费的时间，通过 Thread#sleep(...) 方法，实现重新消费的时间间隔。 注意： FailedRecordTracker 提供的计数是客户端级别的，重启 JVM 应用后，计数是会丢失的。所以，如果想要计数进行持久化，需要自己重新实现下 FailedRecordTracker 类，通过 ZooKeeper 存储计数。 SeekToCurrentErrorHandler 是只针对消息的单条消费失败的消费重试处理。如果想要有消息的批量消费失败的消费重试处理，可以使用 SeekToCurrentBatchErrorHandler 。配置方式如下 @Bean @Primary public BatchErrorHandler kafkaBatchErrorHandler() { // 创建 SeekToCurrentBatchErrorHandler 对象 SeekToCurrentBatchErrorHandler batchErrorHandler = new SeekToCurrentBatchErrorHandler(); // 创建 FixedBackOff 对象 BackOff backOff = new FixedBackOff(10 * 1000L, 3L); batchErrorHandler.setBackOff(backOff); // 返回 return batchErrorHandler; }SeekToCurrentBatchErrorHandler 暂时不支持死信队列的机制。 自定义逻辑处理消费异常支持自定义 ErrorHandler 或 BatchErrorHandler 实现类，实现对消费异常的自定义的逻辑 比如 https://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/springframework/kafka/listener/LoggingErrorHandler.java public class LoggingErrorHandler implements ErrorHandler { private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(LoggingErrorHandler.class)); @Override public void handle(Exception thrownException, ConsumerRecord record) { LOGGER.error(thrownException, () -> "Error while processing: " + ObjectUtils.nullSafeToString(record)); } }配置方式同 SeekToCurrentErrorHandler 或 SeekToCurrentBatchErrorHandler。 生产者 package com.artisan.springkafka.producer; import com.artisan.springkafka.constants.TOPIC; import com.artisan.springkafka.domain.MessageMock; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.SendResult; import org.springframework.stereotype.Component; import org.springframework.util.concurrent.ListenableFuture; import java.util.Random; import java.util.concurrent.ExecutionException; /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/2/17 22:25 * @mark: show me the code , change the world */ @Component public class ArtisanProducerMock { @Autowired private KafkaTemplate kafkaTemplate ; public ListenableFuture sendMsgASync() { // 模拟发送的消息 Integer id = new Random().nextInt(100); MessageMock messageMock = new MessageMock(id,"messageSendByAsync-" + id); // 异步发送消息 ListenableFuture result = kafkaTemplate.send(TOPIC.TOPIC, messageMock); return result ; } }消费者 package com.artisan.springkafka.consumer; import com.artisan.springkafka.domain.MessageMock; import com.artisan.springkafka.constants.TOPIC; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Component; /** * @author 小工匠 * @version 1.0 * @description: TODO * @date 2021/2/17 22:33 * @mark: show me the code , change the world */ @Component public class ArtisanCosumerMock { private Logger logger = LoggerFactory.getLogger(getClass()); private static final String CONSUMER_GROUP_PREFIX = "MOCK-A" ; @KafkaListener(topics = TOPIC.TOPIC ,groupId = CONSUMER_GROUP_PREFIX + TOPIC.TOPIC) public void onMessage(MessageMock messageMock){ logger.info("【接受到消息][线程:{} 消息内容：{}]", Thread.currentThread().getName(), messageMock); // 模拟抛出一次一行 throw new RuntimeException("MOCK Handle Exception Happened"); } }在消费消息时候，抛出一个 RuntimeException 异常，模拟消费失败 单元测试 package com.artisan.springkafka.produceTest; import com.artisan.springkafka.SpringkafkaApplication; import com.artisan.springkafka.producer.ArtisanProducerMock; import org.junit.Test; import org.junit.runner.RunWith; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.kafka.support.SendResult; import org.springframework.test.context.junit4.SpringRunner; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureCallback; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; /** * @author 小工匠 * * @version 1.0 * @description: TODO * @date 2021/2/17 22:40 * @mark: show me the code , change the world */ @RunWith(SpringRunner.class) @SpringBootTest(classes = SpringkafkaApplication.class) public class ProduceMockTest { private Logger logger = LoggerFactory.getLogger(getClass()); @Autowired private ArtisanProducerMock artisanProducerMock; @Test public void testAsynSend() throws ExecutionException, InterruptedException { logger.info("开始发送"); artisanProducerMock.sendMsgASync().addCallback(new ListenableFutureCallback() { @Override public void onFailure(Throwable throwable) { logger.info(" 发送异常{}]]", throwable); } @Override public void onSuccess(SendResult objectObjectSendResult) { logger.info("回调结果 Result = topic:[{}] , partition:[{}], offset:[{}]", objectObjectSendResult.getRecordMetadata().topic(), objectObjectSendResult.getRecordMetadata().partition(), objectObjectSendResult.getRecordMetadata().offset()); } }); // 阻塞等待，保证消费 new CountDownLatch(1).await(); } }测速结果我们把这个日志来梳理一下 2021-02-18 16:18:08.032 INFO 25940 --- [ main] c.a.s.produceTest.ProduceMockTest : 开始发送 2021-02-18 16:18:08.332 INFO 25940 --- [ad | producer-1] c.a.s.produceTest.ProduceMockTest : 回调结果 Result = topic:[C_RT_TOPIC] , partition:[0], offset:[0] 2021-02-18 16:18:08.371 INFO 25940 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=15, name='messageSendByAsync-15'}] 2021-02-18 16:18:18.384 ERROR 25940 --- [ntainer#0-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception ...... ...... ...... 2021-02-18 16:18:18.388 INFO 25940 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=15, name='messageSendByAsync-15'}] 2021-02-18 16:18:28.390 ERROR 25940 --- [ntainer#0-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception ...... ...... ...... 2021-02-18 16:18:28.394 INFO 25940 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=15, name='messageSendByAsync-15'}] 2021-02-18 16:18:38.395 ERROR 25940 --- [ntainer#0-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception ...... ...... ...... 2021-02-18 16:18:38.399 INFO 25940 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容：MessageMock{id=15, name='messageSendByAsync-15'}]清晰了么 老兄？ 是不是和我们设置的消费重试 BackOff backOff = new FixedBackOff(10 * 1000L, 3L);10秒 重试3次 3次处理后依然失败，转入死信队列 看看数据 源码地址https://github.com/yangshangwei/boot2/tree/master/springkafkaRetries |
今日新闻 |
推荐新闻 |
| CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3 |